package org.tao.lightningmq.broker.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.model.QueueModel;
import org.tao.lightningmq.broker.utils.LogFileNameUtil;
import org.tao.lightningmq.broker.utils.PutMessageLock;
import org.tao.lightningmq.broker.utils.UnfairReentrantLock;
import org.tao.lightningmq.common.constants.BrokerConstants;
import org.tao.lightningmq.common.enums.FileSizeUnitEnum;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author lht
 * @date 2025/1/4 - 21:44
 * @description: 对consume queue文件做mmap的类
 */
public class ConsumeQueueMMapFileModel {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumeQueueAppendHandler.class);
    private MappedByteBuffer mappedByteBuffer;
    private ByteBuffer readBuffer;
    private String topic;
    private Integer queueId;
    private String consumeQueueFileName;
    private final PutMessageLock putMessageLock;

    public ConsumeQueueMMapFileModel() {
        putMessageLock = new UnfairReentrantLock();
    }

    /**
     * 加载文件做mmap
     * @param topic 消息主题
     * @param queueId 队列id
     * @param startOffset 开始偏移量
     * @param latestWriteOffset 最新写入的偏移量
     * @param mappedSize 映射的大小
     * @param fileSizeUnitEnum 文件大小单位
     * @throws Exception
     */
    public void loadFileMMap(String topic, Integer queueId, int startOffset, int latestWriteOffset, int mappedSize, FileSizeUnitEnum fileSizeUnitEnum) throws Exception {
        this.topic = topic;
        this.queueId = queueId;
        String filePath = getLatestConsumeQueueFile();
        this.doMMap(filePath, startOffset, latestWriteOffset, mappedSize, fileSizeUnitEnum);
    }

    /**
     * 做mmap操作
     * @param filePath 文件路径
     * @param startOffset 开始偏移量
     * @param latestWriteOffset 最新写入的偏移量
     * @param mappedSize 映射的大小
     * @param fileSizeUnitEnum 文件大小单位
     * @throws IOException
     */
    private void doMMap(String filePath, int startOffset, int latestWriteOffset, int mappedSize, FileSizeUnitEnum fileSizeUnitEnum) throws IOException {
        FileSizeUnitEnum unitEnum = fileSizeUnitEnum == null ? FileSizeUnitEnum.B : fileSizeUnitEnum;
        File file = new File(filePath);
        if (!file.exists()) {
            throw new FileNotFoundException("filePath is " + filePath + " inValid");
        }
        mappedSize *= unitEnum.getSize();
        FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
        this.mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, startOffset, mappedSize);
        this.readBuffer = mappedByteBuffer.slice();
        this.mappedByteBuffer.position(latestWriteOffset);
    }

    private String getLatestConsumeQueueFile() {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            throw new RuntimeException("topic is " + topic + " inValid");
        }
        List<QueueModel> queueList = mqTopicModel.getQueueList();
        QueueModel queueModel = queueList.get(queueId);
        if (queueModel == null) {
            throw new RuntimeException("queueId is " + queueId + " inValid");
        }
        int diff = queueModel.countDiff();
        String filePath = null;
        if (diff > 0) {
            filePath = createNewConsumeQueueFile(queueModel.getFileName());
        } else {
            filePath = LogFileNameUtil.buildConsumeQueueFilePath(topic, queueId, queueModel.getFileName());
        }
        return filePath;
    }

    private String createNewConsumeQueueFile(String fileName) {
        String newFileName = LogFileNameUtil.incrConsumeQueueFileName(fileName);
        String newFilePath = LogFileNameUtil.buildConsumeQueueFilePath(topic, queueId, newFileName);
        try {
            boolean newFile = new File(newFilePath).createNewFile();
            if (newFile) {
                LOGGER.info("create new consume queue file");
            }
            return newFilePath;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void writeContent(byte[] content) {
        writeContent(content, false);
    }

    public void writeContent(byte[] content, boolean force) {
        try {
            putMessageLock.lock();
            mappedByteBuffer.put(content);
            if (force) {
                mappedByteBuffer.force();
            }
        } finally {
            putMessageLock.unlock();
        }
    }

    public byte[] readContent(int pos) {
        ByteBuffer slice = readBuffer.slice();
        slice.position(pos);
        byte[] readBytes = new byte[BrokerConstants.CONSUME_QUEUE_EACH_MSG_SIZE];
        slice.get(readBytes);
        return readBytes;
    }

    public List<byte[]> readContentBatch(int pos, int size) {
        ByteBuffer slice = readBuffer.slice();
        slice.position(pos);
        List<byte[]> res = new ArrayList<>(size);
        for (int i = 0; i < size; i++) {
            byte[] readBytes = new byte[BrokerConstants.CONSUME_QUEUE_EACH_MSG_SIZE];
            slice.get(readBytes);
            res.add(readBytes);
        }
        return res;
    }

    public Integer getQueueId() {
        return queueId;
    }
}
